package com.lxn.until

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

import java.util
import scala.collection.mutable

object MYKafkaUtils {

  //首先，我们这里不是直接使用kafka进行数据消费，我们需要传递给sparkstream里面进行消费

  //创建消费者

  //这里我们创建一个可变Map集合，用来配置kafka
  private val consumerConfigs : mutable.Map[String,String] = mutable.Map[String,String](
    //配置连接集群的主机和端口
    //不要硬编码，这种配置尽量不要在工具类里面出现
//    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop102:9200,hadoop103:9200,hadoop104:9200",
    //然后其实还是存在硬编码问题，我们可以添加一个配置类，用来存放前一个配置类的配置名称，使用一个不变的值代替
//    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> MyPropertyUtils("kafka.bootstrap-server"),
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> MyPropertyUtils(MyConfig.KAFKA_BOOTSTRAP_SERVERS),
    //配置kv反序列化，使对象能够传递  根据生成的数据，这里使用字符反序列化器
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> "org.apache.kafka.common.serialization.StringDeserializer",
    //配置组名 组id
//    ConsumerConfig.GROUP_ID_CONFIG -> "stars",  //这里是因为不要写死了，我们需要传递groupid进行消费
    //配置是否自动提交事务
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> "true",
//    ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG -> "5000"   //设置自动提交的时间间隔，默认单位是毫秒
    //配置自动重置的偏移量，一个是开始，一个是结束位置消费数据
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
  )

  //创建获取消费者
  def getConsumer(ssc: StreamingContext, topic: String, groupId: String): InputDStream[ConsumerRecord[String, String]] = {
    //我们在调用这个函数获取消费者的时候，我们可以指定消费的主题，和消费者组id,就可以动态的使用不同的消费者组消费不同的主题
    //由于之前我们还没有对配置文件中进行组id的设置，我们就可以在这个函数里面使用配置对象配置groupId，这样就实现了动态消费者组
    //我们注意,使用map可变集合配置kafka和使用配置对象的配置方法不一样，前者在（）里面使用配置项 -> 配置信息   这里使用put,前面是配置项，后面是配置信息
    consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    //这里将数据穿到stream，所以我们使用工具类创建消费者，并将消费者返回
    val KafkaDStream: InputDStream[ConsumerRecord[String, String]]
    = KafkaUtils.createDirectStream(ssc: StreamingContext,
      LocationStrategies.PreferConsistent, //这个参数经常使用，可以转到定义查看，另一个参数只有在spark的excutor和kafka在同节点时使用
      ConsumerStrategies.Subscribe[String, String](Array(topic), consumerConfigs)) //这个参数是订阅主题的，可能消费多个主题我们就使用数组，然后还需要我们传入kafka的配置参数，这里我们使用一个可变map集合配置
    //我们知道，kafka里面的数据是key value形式存在的，所以订阅主题的时候要指定kv的类型
    //让stream获取消费者，我们需要返回
    KafkaDStream
  }

  //这里我们需要精准消费，所以我们需要实现根据指定的offset进行消费，我们重载一下这个函数
  //创建获取消费者
  def getConsumer(ssc: StreamingContext, topic: String, groupId: String, offset:Map[TopicPartition,Long] ): InputDStream[ConsumerRecord[String, String]] = {
    //我们在调用这个函数获取消费者的时候，我们可以指定消费的主题，和消费者组id,就可以动态的使用不同的消费者组消费不同的主题
    //由于之前我们还没有对配置文件中进行组id的设置，我们就可以在这个函数里面使用配置对象配置groupId，这样就实现了动态消费者组
    //我们注意,使用map可变集合配置kafka和使用配置对象的配置方法不一样，前者在（）里面使用配置项 -> 配置信息   这里使用put,前面是配置项，后面是配置信息
    consumerConfigs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
    //这里将数据穿到stream，所以我们使用工具类创建消费者，并将消费者返回
    val KafkaDStream: InputDStream[ConsumerRecord[String, String]]
    = KafkaUtils.createDirectStream(ssc: StreamingContext,
      LocationStrategies.PreferConsistent, //这个参数经常使用，可以转到定义查看，另一个参数只有在spark的excutor和kafka在同节点时使用
      ConsumerStrategies.Subscribe[String, String](Array(topic), consumerConfigs, offset)) //这个参数是订阅主题的，可能消费多个主题我们就使用数组，然后还需要我们传入kafka的配置参数，这里我们使用一个可变map集合配置
    //我们知道，kafka里面的数据是key value形式存在的，所以订阅主题的时候要指定kv的类型
    //让stream获取消费者，我们需要返回
    KafkaDStream
  }


  //定义生产者
  val producer = createProducer();

  //创建生产者
  def createProducer(): KafkaProducer[String, String] = {
    //new一个生产者之后，我们需要添加配置信息，这里使用一个hashMap配置
    val producerConfigs: util.HashMap[String, AnyRef] = new util.HashMap[String, AnyRef]();
    //配置主机连接信息
    producerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, MyPropertyUtils(MyConfig.KAFKA_BOOTSTRAP_SERVERS))
    //配置kv序列化
    producerConfigs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    producerConfigs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    //配置acks的值  这个值使用来协调数据读取与数据备份的，这里all是最安全的但也是读取最慢的
    producerConfigs.put(ProducerConfig.ACKS_CONFIG, "all")
    //配置幂等性操作
    producerConfigs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true")
    //除了这些，还有其他的配置参数，有空可以复习一下
    //创建生产者与消费者不同，我们直接new一个生产者
    val kafkaProducer = new KafkaProducer[String, String](producerConfigs)
    //返回生产者
    kafkaProducer
  }

  //发送数据
  def send(topic : String , msg : String): Unit = {
    //首先，我们使用ctrl + p 发现里面需要传一个ProducerRecord对象，所以我们可以new一个，然后跟上参数就好
    producer.send(new ProducerRecord[String,String](topic,msg));
  }
  //函数重载，这样后面方便我们使用key进行分区
  def send(topic : String , key : String , msg : String): Unit = {
    //new ProducerRecord[String,String] 这里面指定的数据类型是topic的数据类型 kv的数据类型
    producer.send(new ProducerRecord[String,String](topic,key,msg))
  }

  //由于这是一个单列对象，所以这个程序只能创建一个对象
  //关闭生产者对象
  def close(): Unit = {
    if(producer != null) producer.close()
  }

  //刷写数据
  def flush(): Unit = {
    producer.flush()
  }
}
